package com.ruyuan.dfs.namenode.editslog;

import com.ruyuan.dfs.namenode.config.NameNodeConfig;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.MappedByteBuffer;
import java.nio.channels.FileChannel;
import java.util.ArrayList;
import java.util.List;

/**
 * @ClassName EditLogMappBuffer
 * @Author kebukeyi
 * @Date 2022/6/3 13:03
 * @Description
 * @Version 1.0.0
 */
@Slf4j
public class EditLogMappBuffer {

    private final NameNodeConfig nameNodeConfig;
    private MappedByteBuffer buffer;
    private volatile long endTxid = 0;
    private volatile long startTxid = 0;
    private String fileName;
    private FileChannel channel;
    private int postition;

    public EditLogMappBuffer(NameNodeConfig nameNodeConfig) {
        this.nameNodeConfig = nameNodeConfig;
        fileName = nameNodeConfig.getEditlogsFile(0, nameNodeConfig.getDEFAULT_EDITLOG_COUNT());
        int size = nameNodeConfig.getDEFAULT_MAPPBUFFER_INIT_SIZE();
        buffer = generateMappBuffer(fileName, 0, size);
    }

    public EditLogMappBuffer(NameNodeConfig nameNodeConfig, long startTxid, long endTxid, int postition) {
        this.nameNodeConfig = nameNodeConfig;
        fileName = nameNodeConfig.getEditlogsFile(startTxid, endTxid);
        int size = nameNodeConfig.getDEFAULT_MAPPBUFFER_INIT_SIZE();
        buffer = generateMappBuffer(fileName, postition, size);
        this.startTxid = startTxid;
        this.endTxid = startTxid;
        this.postition = postition;
    }

    /**
     * 生成一个mappedByteBuffer
     *
     * @param fileName
     * @param position
     * @param size
     * @return
     */
    private MappedByteBuffer generateMappBuffer(String fileName, int position, int size) {
        try {
            close();
            channel = new RandomAccessFile(fileName, "rw").getChannel();
            return channel.map(FileChannel.MapMode.READ_WRITE, position, size);
        } catch (IOException e) {
            e.printStackTrace();
            throw new RuntimeException(e);
        }
    }

    /**
     * 写入一条数据到缓冲区
     *
     * @param editLog editlog
     * @throws IOException IO异常
     */
    public void write(EditLogWrapper editLog) throws IOException {
        if (buffer.remaining() < editLog.toByteArray().length) {
            buffer.force();
            this.postition += buffer.position();
            buffer = generateMappBuffer(fileName, this.postition, editLog.toByteArray().length);
        }
        buffer.put(editLog.toByteArray());
        endTxid++;
    }

    /**
     * 获取当前缓冲区的EditLog
     *
     * @return 当前缓冲区的EditLog
     */
    public List<EditLogWrapper> getCurrentEditLog() {
        int position = buffer.position();
        if (position == 0) {
            return new ArrayList<>();
        }
        byte[] bytes = new byte[position];
        buffer.flip();
        buffer.get(bytes, 0, bytes.length);
        buffer.position(position);
        return EditLogWrapper.parseFrom(bytes);
    }

    /**
     * 清除缓冲区
     */
    public void clear() {
        buffer.reset();
        startTxid = endTxid;

    }

    /**
     * 返回当前缓冲区大小
     */
    public Integer size() {
        return buffer.position();
    }

    /**
     * 返回当前日志记录的条数
     *
     * @return
     */
    public Long count() {
        return endTxid - startTxid;
    }

    /**
     * 刷磁盘
     */
    public EditslogInfo flush() throws IOException {
        buffer.force();
        log.info("保存editslog文件：[file={}]", fileName);
        return new EditslogInfo(startTxid, endTxid, fileName);
    }

    public void close() {
        if (channel != null) {
            try {
                channel.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
